package nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author fengbo
 * @version 1.0.0
 * @date 2018/11/29 14:09
 * @since JDK 1.8
 */
public class NioMultiThreadServer {

    private final int port;
    private final CharsetDecoder decoder;
    private AtomicInteger count = new AtomicInteger(0);
    private Task thread1;
    private Task thread2;

    private NioMultiThreadServer(int port) {
        this.port = port;
        this.decoder = StandardCharsets.UTF_8.newDecoder();
    }

    private void listen() {

        ServerSocketChannel serverChannel;
        Selector selector;
        try {
            // 1. 打开一个服务端Channel
            // Netty在绑定端口时(b.bind(PORT).sync())，会调用AbstractBootstrap的bind(int)方法。
            // 该方法会调用本类的doBind(SocketAddress)方法 –> initAndRegister() –> ReflectiveChannelFactory.newChannel()。
            // ReflectiveChannelFactory对象在调用AbstractBootstrap类的channel(Class<? extends C>)方法时创建。
            // 在newChannel()方法中会根据创建ReflectiveChannelFactory对象时传递进来的Class对象反射构建一个SocketChannel对象。
            // 在NioServerSocketChannel类的构造方法中会调用本类的newSocket(SelectorProvider)方法。
            // newSocket()方法会通过java.nio.channels.spi.SelectorProvider的子类创建JDK底层Channel。
            // 创建TCP参数配置类NioServerSocketChannelConfig。该类负责配置newSocket(SelectorProvider)方法构建的JDK底层Channel。
            // 调用父类AbstractNioChannel的构造方法设置Channel为非阻塞并调用父类AbstractChannel的构造方法。
            serverChannel = ServerSocketChannel.open();
            // 使用非阻塞模式 --> AbstractNioChannel(Channel, SelectableChannel, int)
            serverChannel.configureBlocking(false);
            // NioEventLoop类的构造方法中会调用本类的openSelector()方法，在该方法中会调用JDK底层类SelectorProvider.provider().openSelector()打开一个选择器。
            selector = Selector.open();
            // 2. 注册服务端Channel到Selector
            // AbstractBootstrap的initAndRegister()方法中调用完newChannel()和init(Channel)方法后，会调用EventLoopGroup的register(Channel)方法
            // –> SingleThreadEventLoop类的register(Channel) –> register(ChannelPromise)
            // –> AbstractChannel的内部类AbstractUnsafe的register(EventLoop, ChannelPromise)方法。
            // 在register方法中会首先绑定EventLoop，即绑定线程，表明和该事件进行的一些操作都由该线程完成。
            // 然后该方法会调用本类的register0(ChannelPromise) –> 子类AbstractNioChannel的doRegister()方法。
            // 在doRegister方法中会调用调用java.nio.channels.AbstractSelectableChannel的register方法。
            // 同时在AbstractChannel的register0(ChannelPromise)方法中会调用DefaultChannelPipeline的invokeHandlerAddedIfNeeded()方法。
            // 该方法会负责回调ServerBootstrap类的handler(ChannelHandler)方法绑定的Handler的handlerAdded(ChannelHandlerContext)、channelRegistered(ChannelHandlerContext)等方法。
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);

            // 3. 绑定端口
            // b.bind(int) –> AbstractBootstrap.bind(SocketAddress) –> doBind(SocketAddress)方法中完成初始化和注册Selector后会调用本类的doBind0(ChannelFuture, Channel, SocketAddress, ChannelPromise)方法。
            // 该方法会调用AbstractChannel(NioServerSocketChannel)的bind(SocketAddress localAddress, ChannelPromise promise)方法
            // 然后会调用Pipeline(属于NioServerSocketChannel的Pipeline)的bind方法，然后从Tail节点开始往下传播一直到Head节点
            // Head节点的bind方法会调用AbstractChannel的内部类AbstractUnsafe.bind(SocketAddress, ChannelPromise)
            // –> 子类NioServerSocketChannel.doBind(SocketAddress) –> java.nio.channels.ServerSocketChannel.bind(SocketAddress, int)方法。
            serverChannel.bind(new InetSocketAddress(port));
        } catch (IOException e) {
            e.printStackTrace();
            return;
        }

        while (true) {
            try {
                // 4. 阻塞用户线程并等待连接
                // 在NioEventLoop的run() –> select(boolean oldWakenUp)
                // –> SelectedSelectionKeySetSelector.select(long timeout)
                // 调用JDK底层sun.nio.ch.SelectorImpl.select(long)方法阻塞线程并等待用户请求。
                selector.select();
            } catch (IOException e) {
                e.printStackTrace();
                break;
            }
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> keyIterator = keys.iterator();
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();
                try {
                    if (key.isAcceptable()) {
                        // 5. 检测新连接并创建客户端Channel
                        // 在NioEventLoop的run()中调用完select(boolean)方法后会调用本类的processSelectedKeys()
                        // –> processSelectedKeysOptimized() –> processSelectedKey(SelectionKey, AbstractNioChannel)
                        // –> AbstractNioMessageChannel的内部类NioMessageUnsafe.read()
                        // –> 子类NioServerSocketChannel.doReadMessages(List) –> SocketUtils.accept(ServerSocketChannel)。
                        // 调用NioSocketChannel(Channel, SocketChannel)的构造函数 –> AbstractNioByteChannel(Channel, SelectableChannel)
                        // –> AbstractNioChannel(Channel, SelectableChannel, int) –> AbstractChannel(Channel)。
                        SocketChannel client = ((ServerSocketChannel) key.channel()).accept();
                        // 6. 分配线程及注册Selector
                        // 在AbstractBootstrap类的doBind(final SocketAddress localAddress)会调用本类的initAndRegister()方法
                        // 该方法会调用子类ServerBootstrap的init(Channel)方法
                        // 该方法会构建一个ServerBootstrap的内部类ServerBootstrapAcceptor的实例并添加到Pipeline中
                        // 注册Selector的业务逻辑在该Handler的channelRead(ChannelHandlerContext, Object)方法中。
                        // 该方法中会调用MultithreadEventLoopGroup.register(Channel) –> SingleThreadEventLoop.register(Channel)
                        if (client != null) {
                            if (count.incrementAndGet()%2 == 0) {
                                if (thread1 == null) {
                                    thread1 = new Task("thread1");
                                    thread1.start();
                                }
                                thread1.register(client);
                            } else {
                                if (thread2 == null) {
                                    thread2 = new Task("thread2");
                                    thread2.start();
                                }
                                thread2.register(client);
                            }
                        }
                    }
                } catch (IOException e) {
                    key.cancel();
                    closeChannel(key.channel());
                    e.printStackTrace();
                }
                keyIterator.remove();
            }

        }
    }

    private void closeChannel(Channel channel) {
        try {
            channel.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        NioMultiThreadServer server = new NioMultiThreadServer(8080);
        server.listen();
    }

    class Task extends Thread {

        /** 每个线程都有自己的Selector */
        private Selector selector;

        Task(String name) {
            super(name);
            try {
                this.selector = Selector.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        void register(SocketChannel client) {
            try {
                client.configureBlocking(false);
                // 7. 将客户端channel注册到selector上
                // 在AbstractChannel的内部类AbstractUnsafe的register0(ChannelPromise)方法
                // 调用DefaultChannelPipeline类的fireChannelActive()
                // –> AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext)
                // –> invokeChannelActive() –> DefaultChannelPipeline.channelActive(ChannelHandlerContext)
                // –> readIfIsAutoRead() –> AbstractChannel.read() –> DefaultChannelPipeline.read()
                // –> AbstractChannelHandlerContext.read() –> invokeRead()
                // –> DefaultChannelPipeline.read(ChannelHandlerContext) –> AbstractUnsafe.beginRead() –> AbstractNioChannel.doBeginRead()。
                // 这里会并发操作selector，netty的解决办法是构建一个Queue，将外部任务放到Queue里面
                // 等当前线程不阻塞后自己执行注册到selector的逻辑，而不是外部线程来进行注册
                client.register(selector, SelectionKey.OP_READ);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            while (true) {
                try {
                    // NioEventLoop.run() --> NioEventLoop.select(boolean)
                    selector.select();
                } catch (IOException e) {
                    e.printStackTrace();
                    break;
                }
                // 8. 读取客户的发来的数据
                // 在NioEventLoop的run() –> processSelectedKeys() –> processSelectedKeysOptimized()
                // –> processSelectedKey(SelectionKey, AbstractNioChannel)
                // 调用AbstractNioMessageChannel内部类NioMessageUnsafe或AbstractNioByteChannel的内部类NioByteUnsafe的read()方法
                // –> NioSocketChannel.doReadBytes(ByteBuf) –> AbstractByteBuf.writeBytes(ScatteringByteChannel, int length)
                // –> PooledUnsafeDirectByteBuf.setBytes(int index, ScatteringByteChannel, int length)
                // 调用java.nio.channels.SocketChannel的read(ByteBuffer)方法。
                Set<SelectionKey> keys = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = keys.iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    try {
                        if (key.isReadable()) {
                            SocketChannel client = (SocketChannel) key.channel();
                            ByteBuffer buffer = ByteBuffer.allocate(1024);
                            client.read(buffer);
                            // flip方法将Buffer从写模式切换到读模式
                            buffer.flip();
                            // 接收请求
                            String message = decoder.decode(buffer).toString();
System.out.println(message);
                            // 改变自身关注事件，可以用位或操作|组合时间
                            key.interestOps(SelectionKey.OP_WRITE);
                        } else if (key.isWritable()) {
                            SocketChannel client = (SocketChannel) key.channel();
                            byte[] bytes = Thread.currentThread().getName().getBytes();
                            client.write(ByteBuffer.wrap(bytes));
                            key.interestOps(SelectionKey.OP_READ);
                        }
                    } catch (IOException e) {
                        key.cancel();
                        closeChannel(key.channel());
                    }
                    keyIterator.remove();
                }

            }
        }
    }
}

